package com.example.java.table;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.*;
import org.apache.flink.types.Row;

public class TableFromKafkaToFSMain {
    public static void main(String[] args) throws Exception {
        //Flink执行环境env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //用env，做出Table环境tEnv
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .withBuiltInCatalogName("default_catalog")
                .withBuiltInDatabaseName("default_database")
                .build();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);
        // 声明Kafka连接器
        ConnectTableDescriptor kafkaDescriptor = tEnv.connect(
                new Kafka()
                        .version("universal")
                        .topic("user")
                        .startFromEarliest()
                        .property("bootstrap.servers", "linux122:9092")
        )
                .withFormat(new Csv())
                .withSchema(new Schema().field("name", DataTypes.STRING())
                        .field("age", DataTypes.INT()));

        kafkaDescriptor.createTemporaryTable("myUser");

        Table table = tEnv.sqlQuery("select * from myUser");
        DataStream<Tuple2<Boolean, Row>> result = tEnv.toRetractStream(table, Row.class);
        result.print();

        // 将结果输出到文件中
        tEnv.connect(new FileSystem().path("result\\out.txt"))
                .withFormat(new Csv())
                .withSchema(new Schema().field("name",
                        DataTypes.STRING()).field("age", DataTypes.INT()))
                .createTemporaryTable("outputTable");
        table.executeInsert("outputTable");
        env.execute();
    }
}
